package com.java.schedule.springbatch.job;

import com.java.schedule.springbatch.entity.EmployeeFrom;
import com.java.schedule.springbatch.entity.EmployeeDTO;
import com.java.schedule.springbatch.entity.EmployeeTo;
import com.java.schedule.springbatch.listener.MySecondReadListener;
import com.java.schedule.springbatch.mapper.to.EmployeeToMapper;
import com.java.schedule.springbatch.process.CustomBatchItemWriter;
import com.java.schedule.springbatch.process.EmployeeProcessor;
import org.apache.ibatis.session.SqlSessionFactory;
import org.mybatis.spring.batch.MyBatisBatchItemWriter;
import org.mybatis.spring.batch.MyBatisCursorItemReader;
import org.mybatis.spring.batch.MyBatisPagingItemReader;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.step.tasklet.TaskletStep;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.database.BeanPropertyItemSqlParameterSourceProvider;
import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;

import javax.annotation.Resource;
import javax.sql.DataSource;
import java.util.ArrayList;
import java.util.List;

@Configuration
public class Demo1 {
    @Resource
    private JobBuilderFactory jobBuilderFactory;
    @Resource
    private StepBuilderFactory stepBuilderFactory;
    @Resource
    private EmployeeProcessor employeeProcessor;

    @Resource
    private MySecondReadListener mySecondReadListener;

    @Resource
    private SqlSessionFactory sqlSessionFactory;

    @Resource
    private DataSource dataSource;

    @Resource
    private EmployeeToMapper employeeToMapper;

    @Qualifier(value = "demo4")
    @Bean
    public Job demo4Job() throws Exception {
        return this.jobBuilderFactory.get("demo4")
                .start(step1Demo4())
                .build();
    }

    @Bean
    @StepScope
    public MyBatisPagingItemReader mBatisPagingItemReader(){
        MyBatisPagingItemReader<EmployeeFrom> reader = new MyBatisPagingItemReader<>();
        reader.setSqlSessionFactory(sqlSessionFactory);
        reader.setQueryId("selectByPaging");
        reader.setPageSize(2);
        return reader;
    }

    /**
     * 按照 job 配置的 chunk 值读取条数
     * @return
     */
    @Bean
    @StepScope
    public MyBatisCursorItemReader myBatisCursorItemReader(){
        MyBatisCursorItemReader<EmployeeFrom> reader = new MyBatisCursorItemReader<>();
        reader.setSqlSessionFactory(sqlSessionFactory);
        reader.setQueryId("selectByCursor");
//        reader.setCurrentItemCount(10);
        return reader;
    }

    @Bean
    public Step step1Demo4() throws Exception {
        TaskletStep step1 = this.stepBuilderFactory.get("step1")
                .<EmployeeDTO, EmployeeFrom>chunk(4)
                // 适合大数据量
                .reader(myBatisCursorItemReader())
                // 适合小数据量
//                .reader(mBatisPagingItemReader())
                .processor(employeeProcessor)
//                .writer(customBatchItemWriter())
//                .writer(myBatisBatchItemWriter())
                .writer(jdbcBatchItemWriter())
//                .writer(oneAddMyBatisBatchItemWriter())
//                .writer(forEachWriter())
                // 开启的话会数据重复(age字段查看)
//                .taskExecutor(taskExecutor())
//                .listener(mySecondReadListener)
                .build();
        return step1;
    }


    /**
     * 报错，提示 Unknown error 1136 , 字段数量和值数量不一致
      * @return
     */
//    @Bean
//    public CustomBatchItemWriter<EmployeeTo> customBatchItemWriter() {
//        CustomBatchItemWriter<EmployeeTo> itemWriter = new CustomBatchItemWriter<>();
//        itemWriter.setSqlSessionFactory(sqlSessionFactory);
//        itemWriter.setStatementId("insertByBatch");
//        return itemWriter;
//    }

    /**
     * 通过 <foreach> 标签写入报错
     * @return
     */
    @Bean
    public MyBatisBatchItemWriter<EmployeeTo> myBatisBatchItemWriter() {
        MyBatisBatchItemWriter<EmployeeTo> itemWriter = new MyBatisBatchItemWriter<>();
        itemWriter.setSqlSessionFactory(sqlSessionFactory);
        itemWriter.setStatementId("insertByBatch");
        return itemWriter;
    }

    /**
     * 按照 job 配置的 chunk 值批量写入条数
     * @return
     */
    @Bean
    public MyBatisBatchItemWriter<EmployeeTo> oneAddMyBatisBatchItemWriter() {
        MyBatisBatchItemWriter<EmployeeTo> itemWriter = new MyBatisBatchItemWriter<>();
        itemWriter.setSqlSessionFactory(sqlSessionFactory);
        itemWriter.setStatementId("addEmployee");
        return itemWriter;
    }

    /**
     * 自定义ItemWriter实现批量写入
     * @return
     */
    @Bean
    public ItemWriter<EmployeeTo> forEachWriter() {
        ItemWriter<EmployeeTo> writer = (ItemWriter) list -> {
            //debug发现是嵌套的List reader的线程List嵌套真正的List
            employeeToMapper.insertByBatch(list);
        };
        return writer;
    }

    /**
     * 按照 job 配置的 chunk 值批量写入条数
     * @return
     */
    @Bean
    public JdbcBatchItemWriter<EmployeeTo> jdbcBatchItemWriter() {
        JdbcBatchItemWriter<EmployeeTo> itemWriter = new JdbcBatchItemWriter<>();
        itemWriter.setDataSource(dataSource);
        itemWriter.setSql("INSERT INTO employee_to (employee_id, first_name, last_name, email, age) VALUES (:employeeId, :firstName, :lastName, :email, :age)");
        itemWriter.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>());
        return itemWriter;
    }

    @Bean
    public TaskExecutor taskExecutor(){
        SimpleAsyncTaskExecutor simpleAsyncTaskExecutor = new SimpleAsyncTaskExecutor();
        simpleAsyncTaskExecutor.setConcurrencyLimit(5);
        return simpleAsyncTaskExecutor;
    }



}
